package com.miracle.tears.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @Author: rupert-tears
 * @Date: Created in 18:05 2021/8/31
 * @Description: 优秀的人都在孤独的翻山越岭！
 * java实现离线版 flink计算
 */
public class FlinkWordCount4DataSet {


    public static void main(String[] args) throws Exception {
        // 创建Flink的代码执行离线数据流上下文环境变量
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 定义从本地文件系统当中文件路径
        String filePath;
        if (args == null || args.length == 0) {
            filePath = "S:\\BIGDATA\\项目\\源\\flink";
        } else {
            filePath = args[0];
        }
        // 获取输入文件对应的DataSet对象
        DataSet<String> inputLineDataSet = env.readTextFile(filePath);

        // 对数据集进行多个算子处理，按空白符号分词展开，并转换成(word, 1)二元组进行统计
        DataSet<Tuple2<String, Integer>> resultSet =
                inputLineDataSet
                        .flatMap(
                                new FlatMapFunction<String, Tuple2<String, Integer>>() {
                                    public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                                        // 按空白符号分词
                                        String[] wordArray = line.split("\\s");
                                        // 遍历所有word，包成二元组输出
                                        for (String word : wordArray) {
                                            out.collect(new Tuple2<>(word, 1));
                                        }
                                    }
                                })
                        .groupBy(0) // 返回的是一个一个的(word,1)的二元组，按照第一个位置的word分组
                        .sum(1); // 将第二个位置上的freq=1的数据求和
        // 打印出来计算出来的(word,freq)的统计结果对

        // 注：print会自行执行env.execute方法，故不用再最后执行env.execute正式开启执行过程
        resultSet.print();
        // 注：writeAsText的sink算子，必须要调用env.execute方法才能正式开启环境执行
        // resultSet.writeAsText("d:\\temp\\output2", WriteMode.OVERWRITE)
        // .setParallelism(2);
        // 正式开启执行flink计算
        // env.execute();
    }
}
